Add streaming support to LlmComposer.Agent with :tool_call stream chunks#117
Conversation
…port Introduces StreamCollector, the module responsible for accumulating streaming chunks of a single agent turn and reassembling them into a synthetic LlmResponse identical in shape to a non-streaming one. Supports four providers: - :open_ai / :open_router — index-based fragment merging (identical wire format; OpenRouter delegates to the same logic). - :google — already-complete FunctionCall structs, collected directly. - :bedrock — start events carry toolUseId + name; continuation events carry only inputJson fragments, grouped by toolUseId and concatenated in arrival order via a tool_id_sequence list. Also adds StreamCollector.aggregate_cost_infos/1 for summing per-turn CostInfo structs into a single run-level aggregate.
LlmComposer.Agent.run/3 now supports stream_response: true for all four providers (:open_ai, :open_router, :google, :bedrock). The loop runs intermediate tool-calling turns internally using StreamCollector and forwards only the final answer to the caller token-by-token. The stream ends with a terminal :done chunk carrying cumulative usage/cost_info and the full Agent.Result in metadata.agent_result. Hard failures yield a :error chunk instead. Adds a new :tool_call StreamChunk type: one assembled chunk per executed tool call (with result filled in) is emitted right after execution and before the next LLM turn, making the stream self-contained without needing a telemetry handler. Intermediate progress (reasoning, per-iteration data) continues to be exposed via [:llm_composer, :agent, :*] telemetry events. A new telemetry_metadata option scopes all events to a single run via an auto-generated run_id.
3839ed3 to
eba8d91
Compare
nvelasco
left a comment
There was a problem hiding this comment.
@sonic182 , I don't have too much context yet.
So I was reviewing the changes with claude, here is a list of things that may worth to review it
- to_string(call.result) can crash the whole loop (Agent #111, via function_call_helpers.ex:53)
Tool results are fed back with to_string/1. A tool returning a map/list/struct (common) raises Protocol.UndefinedError and aborts
both sequential and parallel runs — contradicting the "tool failures are fed back, not aborted" promise. Confirmed untested: every
test tool returns only numbers/strings. → JSON-encode non-binary results. - :max_iterations_reached discards all accumulated work (Agent #111, agent.ex loop/4)
Returns a bare {:error, :max_iterations_reached}, throwing away cost_infos, function_calls, and the partial conversation — caller
can't see what was billed. → Return a partial Result or attach accumulated cost to the error. - Guide provider list is incomplete (#113, guides/agent.md)
Claims "OpenAI, OpenRouter, Google," but Bedrock (#98) and Ollama also support function calling and the loop is provider-agnostic. →
Say "any function-calling provider"; don't omit Bedrock/Ollama.
Worth a conscious confirmation
- Behavior change: Keyword.fetch!(:model) → Keyword.get(:model) (telemetry #112, providers_runner.ex)
Old code raised on a missing :model; new code silently yields nil in telemetry/metrics. Removes a misconfiguration guard. - functions_from_settings/1 only reads the first provider (Agent #111, agent.ex)
In a multi-provider fallback setup with per-provider functions, the default tool set can mismatch the provider actually selected at
runtime.
Minor / cosmetic
- Telemetry fires from child processes — execute_one's :telemetry.span runs inside Task.async_stream tasks (parallel mode); caller
Logger/telemetry_metadata won't auto-propagate. - :agent, :iteration, :stop has no matching :start/duration, and for tool iterations fires after tool execution — slightly
asymmetric vs. the span-based events. - O(n) list appends each iteration (cost_infos ++, messages ++, function_calls ++) — negligible at default max_iterations: 10.
- Fix to_string/1 crash in FunctionCallHelpers when tool results are maps/structs; JSON-encode non-binary results via Helpers.json_engine() with inspect/1 fallback - Return partial result (cost_infos, function_calls, messages) on :max_iterations_reached instead of discarding accumulated work; both sync and streaming paths updated - Fix functions_from_settings/1 to merge functions from all providers in multi-provider fallback setups, not just the first one - Propagate Logger.metadata into Task.async_stream workers in parallel tool execution mode - Add intentional comment on Keyword.get(:model) in providers_runner - Expand StreamCollector to support :open_ai_responses and :ollama; add merge_tool_calls/to_function_calls clauses for Responses API tool-call streaming, handling the missing call_id in delta events via current_tool_id fallback - Inject additionalProperties: false into OpenAI Responses tool schemas when strict mode is active - Log response body on non-200 errors in OpenAI Responses provider
nvelasco
left a comment
There was a problem hiding this comment.
What about point 3 ?
- Incomplete provider list in the guide — Not addressed. The commit
updated the moduledocs in agent.ex and stream_collector.ex (now say
"all providers" + Ollama caveat), but guides/agent.md was never
touched. At the PR head it still reads:
▎ line 13–15: "The loop is synchronous (streaming is not supported
▎ in this version)* … works with any provider that supports function
▎ calling (OpenAI, OpenRouter, Google)."*
▎ line 95: "Streaming settings (stream_response: true) return
▎ {:error, :streaming_not_supported}."
This is now doubly stale: it still omits Bedrock/Ollama and claims
streaming is unsupported — directly contradicting what this entire
PR adds. Worth asking @sonic182 to update guides/agent.md to match
the new moduledocs (and the new streaming.md guide).
guide update (docs) is for pr3, the nex of this for usage struct, we already have a |
PR Series
:tool_callstream chunksSummary
Streaming agent loop:
LlmComposer.Agent.run/3now returns{:ok, stream}whensettings.stream_responseistrue. The stream yields only the final answer's:text_deltachunks followed by a terminal:donechunk with cumulativeusage/cost_infoand the fullAgent.Resultinmetadata.agent_result. Intermediate tool-calling turns run internally viaStreamCollector(PR Add LlmComposer.Agent.StreamCollector (OpenAI, OpenRouter, Google, Bedrock) #116). Supported providers::open_ai,:open_router,:open_ai_responses,:google,:bedrock,:ollama(native Ollama does not emit tool-call deltas; use:open_aipointed at Ollama's OpenAI-compatible endpoint for tool-call streaming).:tool_callstream chunk type: After executing tool calls the agent emits one assembled%StreamChunk{type: :tool_call}chunk per call (result already filled in) before starting the next LLM turn. This makes the stream self-contained — callers can react to tool activity inline without attaching a telemetry handler:Telemetry:
[:llm_composer, :agent, :tool, :start]now includes:arguments,:metadata, and:id. A new[:llm_composer, :agent, :reasoning, :delta]event surfaces intermediate reasoning. Passtelemetry_metadata:torun/3to scope all events to a single run via an auto-generated:run_id.Dialyzer fix: removed unreachable fallback clause in
user_prompt/2(Settings.user_prompt_prefixis always aString.t()).Bug fixes
Tool result encoding crash:
FunctionCallHelpers.build_tool_result_messages/1usedto_string/1on tool results, raisingProtocol.UndefinedErrorfor any tool returning a map, list, or struct. Results are now JSON-encoded viaHelpers.json_engine()with aninspect/1fallback.:max_iterations_reacheddiscards accumulated work: the loop returned a bare{:error, :max_iterations_reached}, throwing awaycost_infos,function_calls, and the partial conversation. Now returns{:error, {:max_iterations_reached, %{cost_infos, function_calls, messages, iterations}}}. The streaming path includes the same data in the error chunk metadata.functions_from_settings/1only read the first provider: in a multi-provider fallback setup, only the first provider's:functionswere used. Now merges functions from all providers and deduplicates by name.Logger metadata not propagated in parallel tool execution:
Task.async_streamworkers in:parallelmode didn't inherit the caller'sLogger.metadata/0. Parent metadata is now captured and restored inside each worker.Test plan
mix test test/llm_composer/agent_test.exsmix precommitsamples/loop_agent_all_providers_stream.ex